package org.ala.tiktools.rpc.plugin.serializer.kryo;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

import org.ala.tiktools.rpc.config.RpcConfigInfo;
import org.ala.tiktools.rpc.exception.RpcRuntimeException;
import org.ala.tiktools.rpc.plugin.serializer.ISerializer;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.ByteBufferOutput;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.KryoBufferOverflowException;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.serializers.CompatibleFieldSerializer;


/**
 * 与线程环境绑定的kryo
 * <p>  彻底无锁
 * <p>  但只能在特定线程中使用，比如EventLoop，而且线程会比较重
 *
 * @author ala
 * @date 2025-04-04 15:47
 */
public class KryoThreadSerializer implements ISerializer {
	
	
	public static KryoThreadSerializer INSTANCE = new KryoThreadSerializer();
	
	
    /**
     * 线程变量管理
     */
    static ThreadLocal<Kryo> kryoTL = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        //  不强制要求每个类必须注册过
        kryo.setRegistrationRequired(false);
        //  检测循环引用时只保留第一个对象，而不是每个循环引用都new一个
        kryo.setReferences(true);
        //	设置默认的序列化反序列化器，兼容字段的增删改。但性能会有所下降
        kryo.setDefaultSerializer(CompatibleFieldSerializer.class);
        
        //  注册每个消息体
//        kryo.register(RpcPushBody.class);
//        kryo.register(RpcLoginBody.class);
//        kryo.register(RpcAckBody.class);
//        kryo.register(java.lang.Object[].class);
        return kryo;
    });
    static ThreadLocal<ByteBufferOutput> outputTL = new ThreadLocal<>();
    private static ByteBufferOutput getOutput() {
    	ByteBufferOutput out = outputTL.get();
    	if (out == null) {
    		int size = RpcConfigInfo.INSTANCE.getSerializeDefSize();
    		if (size <= 0) {size = 4096;}
    		out = new ByteBufferOutput(size);
    		outputTL.set(out);
    	}
    	return out;
    }
    static ThreadLocal<Input> inputTL = new ThreadLocal<>();
    private static Input getInput() {
    	Input in = inputTL.get();
    	if (in == null) {
    		int size = RpcConfigInfo.INSTANCE.getSerializeDefSize();
    		if (size <= 0) {size = 4096;}
    		in = new Input(size);
    	}
    	return in;
    }
    

    @Override
    public byte[] serialize(Object o) {
        ByteBufferOutput out = getOutput();
        try {
            kryoTL.get().writeObject(out, o);
            return out.toBytes();
        } catch (Exception e) {
        	//	如果是buffer超限了，用临时buffer
        	if (e instanceof KryoBufferOverflowException) {
        		return serializeByTempBuffer(o);
        	}
        	throw new RpcRuntimeException(e);
        } finally {
            out.reset();
        }
    }
    /**
     * 	用临时buffer
     */
    public byte[] serializeByTempBuffer(Object o) {
    	ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
	    Output out = new Output(outputStream);
	    try {
	    	kryoTL.get().writeObject(out, o);
	        out.flush();
	        return outputStream.toByteArray();
	    } finally {
	    	out.close();
	    }
    }

    @Override
    public <T> T deSerialize(byte[] bytes, Class<T> clazz) {
    	if (bytes.length > RpcConfigInfo.INSTANCE.getSerializeDefSize()) {
    		return deSerializeByTempBuffer(bytes, clazz);
    	}
        Input in = getInput();
        try {
            in.setBuffer(bytes);
            return kryoTL.get().readObject(in, clazz);
        } finally {
            in.reset();
        }
    }
    /**
     * 	用临时buffer
     */
    public <T> T deSerializeByTempBuffer(byte[] bytes, Class<T> clazz) {
    	ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
    	Input in = new Input(byteArrayInputStream);
    	try {
    		return kryoTL.get().readObject(in, clazz);
    	} finally {
    		in.close();
    	}
    }
}
